package org.zachary.commonservice.massage_queue;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.zachary.commonmodule.entity.MqTransmissionJson;
import org.zachary.commonmodule.entity.entity.CreditDimension;
import org.zachary.commonmodule.entity.entity.CreditDimensionDTO;
import org.zachary.commonmodule.entity.entity.KafkaDTO;
import org.zachary.commonmodule.entity.entity.RecordDTO;
import org.zachary.commonmodule.entity.utils.JsonUtil;
import org.zachary.commonservice.controller.OverduePushController;
import org.zachary.commonservice.entity.DimensionDto;
import org.zachary.commonservice.entity.OverdueRecord;
import org.zachary.commonservice.entity.ServiceType;
import org.zachary.commonservice.mapper.CreditScoreStrategyMapper;
import org.zachary.commonservice.mapper.GetUserCreditScoreMapper;
import org.zachary.commonservice.mapper.OverdueMapper;
import org.zachary.commonservice.service.OverdueService;
import org.zachary.commonservice.service.feign.NotifyCreditScoreManager;
import org.zachary.commonservice.strategy.CreditStrategy;
import org.zachary.commonservice.strategy.factory.StrategyFactory;

import java.time.*;

@Slf4j
@Component
public class KafkaConsumer {
    @Autowired
    private  GetUserCreditScoreMapper getUserCreditScoreMapper;
    @Autowired
    private  StrategyFactory strategyFactory;
    @Autowired
    private OverdueService overdueService;
    @Autowired
    private OverduePushController pushController;
    private OverdueMapper overdueMapper;
    @Autowired
    private NotifyCreditScoreManager notifyCreditScoreManager;
    @KafkaListener(topics = "TestTopic", groupId = "gateway-group")
    public void listenCreditScoreUpdate(
            @Payload String message,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            Acknowledgment ack
    ) {
        log.info("信用分计算服务：分区 {} 接收到消息{}", partition, message);
        MqTransmissionJson mq = JsonUtil.deserialize(message); // 反序列化

        if(mq.getIsOverdue() && mq.getOverduePeriod()==-1 && mq.getAmount()<=0.0 && mq.getFinishTime()==null){
            log.info("服务{}设置逾期未付款",matchServiceType(mq.getServiceType()));
            OverdueRecord record=new OverdueRecord(
                    mq.getUuid(),
                    mq.getUserId(),
                    mq.getServiceType(),
                    mq.getIsOverdue(),
                    mq.getOverduePeriod(),
                    mq.getIsPaid(),
                    mq.getAmount(),
                    mq.getDesc(),
                    mq.getDueTime(),
                    mq.getFinishTime()
            );
            overdueService.storeOverdueRecord(record);

        }else{
            try {
                if(mq.getIsOverdue()){
                    OverdueRecord record=new OverdueRecord(
                            mq.getUuid(),
                            mq.getUserId(),
                            mq.getServiceType(),
                            mq.getIsOverdue(),
                            mq.getOverduePeriod(),
                            mq.getIsPaid(),
                            mq.getAmount(),
                            mq.getDesc(),
                            mq.getDueTime(),
                            mq.getFinishTime()
                    );
                    log.info("服务消息，{}，该任务设置为已完成",mq.getDesc());
                    overdueService.setFinished(record);
                    log.info("Kafka完成消息推送：");
                    pushController.pushOverdueUpdate(record);
                }
                // 1. 获取用户当前信用分
                int currentScore = getUserCreditScoreMapper.getUserCreditScore(mq.getUserId());
                // 2. 获取策略
                ServiceType serviceType = ServiceType.values()[mq.getServiceType()];

                CreditStrategy strategy = strategyFactory.getStrategy(serviceType, currentScore);
                // 3. 计算信用分变化
                int scoreDelta = strategy.calculate(mq, currentScore);

                KafkaDTO dto=new KafkaDTO(
                        new CreditDimensionDTO(
                                mq.getUserId(),
                                matchServiceType(mq.getServiceType()),
                                scoreDelta
                        ),
                        new RecordDTO(
                                mq.getUserId(),
                                matchServiceType(mq.getServiceType()),
                                mq.getDesc(),
                                mq.getAmount(),
                                LocalDateTime.ofInstant(mq.getFinishTime(), ZoneId.systemDefault()),
                                matchStatus(mq.getIsOverdue(),mq.getIsPaid(),mq.getAmount(), LocalDateTime.ofInstant(mq.getFinishTime(), ZoneId.systemDefault())),
                                scoreDelta
                        )
                );
                log.info("OpenFeign调用信用管理模块");
                notifyCreditScoreManager.updateScore(dto);
                log.info("OpenFeign调用成功");


                ack.acknowledge();
            } catch (Exception e) {
                log.error("消息处理失败: userId={}, error={}", mq.getUserId(), e.getMessage());
                throw e;
            }
        }
    }
    private static String matchServiceType(Integer serviceTypeId) {
        return switch (serviceTypeId) {
            case 0-> "基本信息";
            case 1,2,3->"亲社会行为";
            case 4,5, 6 -> "信用商业";

            case 7,8, 9 -> "信用生活";

            default -> {
                log.error("不存在此类型服务: {}", serviceTypeId);
                throw new RuntimeException("不存在此类型的服务");
            }
        };
    }
    private static String matchStatus(boolean isOverdue,boolean isPaid,double amount,LocalDateTime finishTime) {
        if(isOverdue && amount<0.0) {
            return "已逾期"+(Duration.between(finishTime, LocalDateTime.now()));
        }
        return "已完成";
    }
}
